package org.hgf.testrabbitmq.consumer.handler;

import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.hgf.testrabbitmq.utils.RedisService;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareMessageListener;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.IOException;
import java.util.concurrent.TimeUnit;

/**
 * 消费者统一处理消息处理类
 *
 * @author 何国发
 * @version 1.0.0
 * @date 2022/9/7 18:46
 */
@Slf4j
@Component
public class ConsumerUnityHandleMessage implements ChannelAwareMessageListener {

    @Resource
    private RedisService redisService;

    @Override
    public void onMessage(Message message, Channel channel) {
        if ("test1".equals(message.getMessageProperties().getConsumerQueue()))
        {
            log.info("=========================================== test1收到消息开始处理.... ==================================================");
            try {
                log.info("消息第一次消费");
                log.info("test1消息ID：{}", message.getMessageProperties().getMessageId());
                int i = 1/0;
                this.ack(message, channel, false);
            } catch (Exception e)
            {
                log.error("消费出现异常：{}", e.getMessage());
                // 消费错误，通知mq进行重发
                this.ack(message, channel, true);
            }
        } else if ("test2".equals(message.getMessageProperties().getConsumerQueue()))
        {
            log.info("======================================== test2收到消息开始处理.... =============================================");
            try {
                String cacheKey = "fanoutExchange:"+message.getMessageProperties().getMessageId();
                Object object = redisService.getCacheObject(cacheKey);
                if (object == null)
                {
                    log.info("消息第一次消费");
                    log.info("test2消息ID：{}", message.getMessageProperties().getMessageId());
                    redisService.setCacheObject(cacheKey, "lock", 30L, TimeUnit.SECONDS);
                } else {
                    log.info("重复消费");
                }
                // 手动ack
                this.ack(message, channel, false);
            } catch (Exception e)
            {
                log.error("消费出现异常：{}", e.getMessage());
                // 消费错误，通知mq进行重发
                this.ack(message, channel, true);
            }
        }
    }

    /**
     * 手动ack
     *
     * @param message   消息对象
     * @param channel   信道对象
     * @param isReject  是否拒绝
     * @author 何国发
     * @date 2022/9/13 16:32
     */
    private void ack(Message message, Channel channel, boolean isReject) {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        // 手动ack
        try {
            // true:将一次性ack所有小于deliveryTag(每次接收消息都会加1)的消息。false，只ack当前消息
            if (isReject)
            {
                channel.basicReject(deliveryTag, false);
            } else {
                channel.basicAck(deliveryTag, false);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
